package com.sh.data.engine.mqtt;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.nio.charset.StandardCharsets;

@Slf4j
@Data
public class EngineMqttClient {

    private IMqttClient mqttClient;

    private String clientId;

    public void init(String host) {

        // MQTT clientId长度有限制，为兼容不同版本，最好长度在23个字符以内
        // https://blog.csdn.net/tiantiandjava/article/details/44154047
        clientId = getRandomClientId();
        // 数据持久化方式：内存
        try {
            mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
        } catch (MqttException e) {
            log.error("clientId {}, 初始化MQTT客户端失败，host {}", clientId, host, e);
        }
    }

    /**
     * 连接服务端broker
     */
    public void connect(MqttCallback callback) {
        String username = "";
        String password = "";
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        // 自动重连
        options.setAutomaticReconnect(true);
        // 是否临时会话
        options.setCleanSession(true);
        // 超时时间
        options.setConnectionTimeout(10);
        options.setKeepAliveInterval(5);
        // 方法回调
        mqttClient.setCallback(callback);
        try {
            mqttClient.connect(options);
        } catch (MqttException e) {
            log.error("clientId {}, 连接MQTT服务器失败", clientId, e);
        }
    }

    /**
     * 断开连接
     */
    public void disConnect() {
        try {
            mqttClient.disconnect();
        } catch (MqttException e) {
            log.error("clientId {}, MQTT客户端断开连接异常", clientId, e);
        }
    }

    /**
     * 断开重连
     */
    public void reConnect() {
        try {
            mqttClient.reconnect();
        } catch (MqttException e) {
            log.error("clientId {}, 重新连接服务端失败", clientId, e);
        }
    }

    /**
     * 发布消息
     *
     * @param topic   主题
     * @param message 消息信息
     * @param qos     消息质量
     * @param retain  是否保留
     */
    public void publishMessage(String topic, String message, int qos, boolean retain) {
        // 发布消息
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setPayload(message.getBytes(StandardCharsets.UTF_8));
        mqttMessage.setQos(qos);
        mqttMessage.setRetained(retain);
        try {
            mqttClient.publish(topic, mqttMessage);
        } catch (MqttException e) {
            log.error(
                "clientId {}, 发布消息失败,topic {},msg {},qos {},retain {}",
                clientId,
                topic,
                message,
                qos,
                retain,
                e);
        }
    }

    /**
     * 订阅
     *
     * @param topicFilter 订阅的主题（带通配符）
     * @param qos
     */
    public void subscribe(String topicFilter, int qos) {
        try {
            mqttClient.subscribe(topicFilter, qos);
            log.info("clientId {},topic {}订阅成功", clientId, topicFilter);
        } catch (MqttException e) {
            log.error("clientId {}, 订阅主题失败", clientId, e);
        }
    }

    /**
     * 取消订阅
     *
     * @param topicFilter
     */
    public void unSubscribe(String topicFilter) {
        try {
            mqttClient.unsubscribe(topicFilter);
        } catch (MqttException e) {
            log.error("clientId {}, 取消订阅失败", clientId, e);
        }
    }

    public String getClientId() {
        return this.clientId;
    }

    private static String getRandomClientId() {
        return "engine_mqtt_" + RandomStringUtils.randomAlphanumeric(10);
    }
}
